package cn.broadsense.handler;

import cn.broadsense.biz.domain.Channel;
import cn.broadsense.biz.domain.ChannelOrg;
import cn.broadsense.biz.dto.ChannelOrgCountDto;
import cn.broadsense.biz.service.ChannelOrgService;
import cn.broadsense.biz.service.ChannelService;
import cn.broadsense.common.core.domain.BaseEntity;
import cn.broadsense.common.utils.spring.SpringUtils;
import cn.broadsense.convert.ChannelConvertSync;
import cn.broadsense.thrid.api.ChannelRepository;
import cn.broadsense.thrid.api.DaHuaApiRepository;
import cn.broadsense.thrid.param.ChannelPageParam;
import cn.broadsense.thrid.param.ChannelStatusParam;
import cn.broadsense.thrid.param.NewChannelParam;
import cn.broadsense.thrid.result.ChannelPageResult;
import cn.broadsense.thrid.result.ChannelResult;
import cn.broadsense.thrid.result.ChannelStatusResult;
import cn.broadsense.thrid.result.NewChannelResult;
import cn.broadsense.xxl.job.core.context.XxlJobHelper;
import cn.broadsense.xxl.job.core.handler.annotation.XxlJob;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.StrUtil;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;
import java.util.*;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.stream.Collectors;

/**
 * @author: pengxin
 * @create: 2023-09-14 10:31
 * @Version 1.0
 **/
@Component
public class DataPullDaHuaJob {

    private static Logger logger = LoggerFactory.getLogger(DataPullDaHuaJob.class);


    @Resource
    private DaHuaApiRepository daHuaApiRepository;

    @Resource
    private ChannelService channelService;

    @Resource
    private ChannelConvertSync channelConvert;

    @Resource
    private ChannelOrgService channelOrgService;

    @Resource
    private ChannelRepository channelRepository;

    final String STAGE_NAME = "五期";


    private final ExecutorService executorService = ThreadUtil.newFixedExecutor(10, 1024, "pullChannelDataV2", true);
    private static final Map<String, String> aMap = initializeAMap();
    private static final Map<Long, String> deptMap = initializeDeptMap();
    private static final Map<String, String> map = initializeMap();

    private static Map<String, String> initializeAMap() {
        Map<String, String> map = new HashMap<>();
        map.put("0", "无功能");
        map.put("1", "支持鱼眼矫正");
        map.put("2", "支持电动聚集");
        map.put("6", "支持红外测温");
        map.put("7", "设备AR");
        map.put("8", "人脸抓拍");
        map.put("9", "车辆抓拍");
        map.put("10", "人车混合抓拍");
        map.put("11", "支持客流功能");
        map.put("12", "目标识别(在人脸抓拍的基础上，还额外具备识别功能)");
        map.put("13", "人体检测(能识别到人体，不保证能抓拍到人脸)");
        map.put("14", "车底扫描");
        map.put("15", "大码率(分辨率超高)");
        map.put("16", "平台AR");
        map.put("17", "人群分布");
        map.put("19", "流量功能");
        map.put("20", "雷达功能");
        return map;
    }

    private static Map<Long, String> initializeDeptMap() {
        Map<Long, String> map = new HashMap<>();
        map.put(101L, "岳麓区");
        map.put(102L, "天心区");
        map.put(200L, "雨花区");
        map.put(201L, "望城区");
        map.put(202L, "开福区");
        map.put(203L, "长沙县");
        map.put(204L, "浏阳市");
        map.put(205L, "宁乡市");
        map.put(206L, "芙蓉区");
        map.put(207L, "高新区");
        return map;
    }

    private static Map<String, String> initializeMap() {
        Map<String, String> map = new HashMap<>();
        map.put("1", "Dahua2");
        map.put("2", "HIKVISION");
        map.put("3", "GB28181协议，使用 RTP over PS 视频流");
        map.put("5", "Dahua-Reg");
        map.put("6", "HIK-Reg");
        map.put("7", "FacePri");
        map.put("8", "BX-5K");
        map.put("9", "ListenV5");
        map.put("10", "FSU");
        map.put("11", "echange");
        map.put("12", "视Imos(智能交通)");
        map.put("13", "abbitMQ");
        map.put("14", "VIoT宇视科技智慧物联");
        map.put("39", "B28059协议，使用 RTP over ES 视频流");
        map.put("99", "nvif");
        map.put("100", "Dh3rdITC");
        map.put("101", "VideoImage GA1400公安视图库协议");
        map.put("103", "DHTrialHost审讯主机");
        map.put("105", "宇视SDK IP方式添加");
        map.put("106", "天地伟业");
        map.put("107", "华为");
        map.put("108", "RTSP");
        map.put("109", "平台Imos");
        map.put("110", "GA/WA3011协议(公共场所无线上网安全管理系统接入规范)");
        map.put("111", "KDIPC");
        map.put("112", "HWIVS");
        map.put("113", "ROSEEK");
        map.put("131", "KMTL");
        map.put("133", "APP_REG 手机APP协议");
        map.put("134", "DB33 地标DB33");
        map.put("135", "JTT808 道路运输车辆卫星定位系统终端通讯协议");
        map.put("136", "DB3311-Reg丽水地标(主动注册)");
        map.put("137", "NOVAJT诺瓦交通协议");
        map.put("138", "DB3311 丽水地标(IP注册)");
        return map;
    }

    /**
     * 1、简单任务示例（Bean模式）
     */
    @XxlJob("pullChannelJobHandler")
    public void pullChannelJobHandler() {
        XxlJobHelper.log("pullChannelJobHandler 开始执行.....");

        String jobParam = XxlJobHelper.getJobParam();
        //默认长沙市
        if (StrUtil.isBlank(jobParam)) {
            jobParam = "10090041757731008";
        }
        SpringUtils.getBean(this.getClass()).pullChannelDataV2(jobParam);
        XxlJobHelper.log("pullChannelJobHandler 执行结束.....");
    }

    @XxlJob("syncChannelStatus")
    public void syncChannelStatus() {
        XxlJobHelper.log("syncChannelStatus 开始执行.....");
        SpringUtils.getBean(this.getClass()).syncDhChannelStatus();
        XxlJobHelper.log("syncChannelStatus 执行结束.....");
    }


    public void syncDhChannelStatus() {
        List<Channel> channelList = channelService.list(Wrappers.lambdaQuery(Channel.class).eq(Channel::getDelFlag,"0").isNotNull(Channel::getOrgCode));
        if (CollUtil.isEmpty(channelList)) {
            return;
        }
        Map<String, List<Channel>> listMap = channelList.stream().collect(Collectors.groupingBy(Channel::getOrgCode));
        CountDownLatch countDownLatch = new CountDownLatch(listMap.size());
        listMap.forEach((orgCode, list) -> {
            executorService.submit(() -> {
                try {

                    ChannelStatusParam statusParam = new ChannelStatusParam();
                    List<ChannelStatusParam.OrgsStructResult> orgs = new ArrayList<>();
                    ChannelStatusParam.OrgsStructResult structResult = new ChannelStatusParam.OrgsStructResult();
                    structResult.setOrgCode(orgCode);
                    List<ChannelStatusParam.OrgsStructResult.ChannelsStructResult> channels = list.stream()
                            .map(channel -> {
                                ChannelStatusParam.OrgsStructResult.ChannelsStructResult channelsStructResult = new ChannelStatusParam.OrgsStructResult.ChannelsStructResult();
                                channelsStructResult.setCode(channel.getChannelCode());
                                return channelsStructResult;
                            })
                            .collect(Collectors.toList());
                    structResult.setChannels(channels);
                    orgs.add(structResult);
                    statusParam.setOrgs(orgs);

                    ChannelStatusResult channelStatusResult = channelRepository.channelStatus(statusParam);
                    list.forEach(e -> e.setStatus(0));
                    if (channelStatusResult != null && CollUtil.isNotEmpty(channelStatusResult.getOrgs())) {
                        List<ChannelStatusResult.OrgsStructResult.ChannelsStructResult> channelsStructResultList = channelStatusResult.getOrgs().get(0).getChannels();
                        if (CollUtil.isNotEmpty(channelsStructResultList)) {
                            //key=code v=status
                            Map<String, String> statusMap = channelsStructResultList.stream().collect(Collectors.toMap(ChannelStatusResult.OrgsStructResult.ChannelsStructResult::getCode, ChannelStatusResult.OrgsStructResult.ChannelsStructResult::getStatus));

                            list.forEach(channel -> {
                                channel.setStatus((statusMap.get(channel.getChannelCode()) != null && "0".equals(statusMap.get(channel.getChannelCode())) ? 1 : 0));
                            });
                        }
                    }

                    //更新数据状态
                    Map<Integer, List<Channel>> syncMap = list.stream().collect(Collectors.groupingBy(Channel::getStatus));
                    List<Channel> offLine = syncMap.get(0);
                    List<Channel> onLine = syncMap.get(1);
                    if (CollUtil.isNotEmpty(offLine)) {
                        channelService.update(Wrappers.lambdaUpdate(Channel.class)
                                .set(BaseEntity::getUpdateTime, new Date())
                                .set(BaseEntity::getUpdateBy, "sys-status")
                                .set(Channel::getStatus, 0).in(Channel::getId, offLine.stream().map(Channel::getId).collect(Collectors.toList())));
                    }

                    if (CollUtil.isNotEmpty(onLine)) {
                        channelService.update(Wrappers.lambdaUpdate(Channel.class)
                                .set(BaseEntity::getUpdateTime, new Date())
                                .set(BaseEntity::getUpdateBy, "sys-status")
                                .set(Channel::getStatus, 1).in(Channel::getId, onLine.stream().map(Channel::getId).collect(Collectors.toList())));
                    }

                } catch (Exception e) {
                    logger.error("同步状态失败：" + e.getMessage());
                    e.printStackTrace();
                } finally {
                    countDownLatch.countDown();
                }

            });
        });

        try {
            countDownLatch.await();
            //同步点位数量
            this.countByOrgQty();
        } catch (InterruptedException e) {
            XxlJobHelper.log("执行失败：{}" + e.getMessage());
            e.printStackTrace();
        }

    }

    @Transactional(rollbackFor = Exception.class)
    public void pullChannelDataV2(String orgCode) {
        Date date = new Date();
        List<ChannelOrg> list = channelOrgService.lambdaQuery()
                .like(StrUtil.isNotBlank(orgCode), ChannelOrg::getFullOrgCode, orgCode).list();
        Map<String, ChannelOrg> orgAllMap = list.stream().collect(Collectors.toMap(ChannelOrg::getOrgCode, e -> e));

        Map<String, ChannelOrg> channelOrgMap = list.stream().filter(e -> e.getDeptId() != null).collect(Collectors.toMap(ChannelOrg::getOrgCode, e -> e));
        int size = list.size();
        CountDownLatch countDownLatch = new CountDownLatch(list.size());
        List<NewChannelResult.AlreadyDealOrgsResult> alreadyDealOrgs = new ArrayList<>();

        for (int i = 0; i < size; i++) {
            ChannelOrg channelOrg = list.get(i);
            executorService.execute(() -> {
                try {
                    NewChannelParam parma = new NewChannelParam();
                    parma.setLayer(channelOrg.getLayer());
                    parma.setOrgCodes(CollUtil.newArrayList(channelOrg.getOrgCode()));
                    NewChannelResult newChannelResults = channelRepository.channelList(parma);
                    if (CollUtil.isNotEmpty(newChannelResults.getAlreadyDealOrgs())) {

                        alreadyDealOrgs.addAll(newChannelResults.getAlreadyDealOrgs());
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                    logger.error("拉取点位失败", e);
                } finally {
                    countDownLatch.countDown();
                }
            });
        }
        try {
            countDownLatch.await();
            if (CollUtil.isEmpty(alreadyDealOrgs)) {
                XxlJobHelper.log("无点位数据");
                return;
            }
            List<Channel> channelList = new ArrayList<>();
            for (NewChannelResult.AlreadyDealOrgsResult result : alreadyDealOrgs) {
                if (result == null) {
                    continue;
                }
                String orgCode2 = result.getOrgCode();
                ChannelOrg channelOrg = orgAllMap.get(orgCode2);
                String fullOrgCode = channelOrg.getFullOrgCode();
                Long deptId = channelOrg.getDeptId();
                String deptName = null;
                Optional<String> first = channelOrgMap.keySet().stream().filter(e -> e.indexOf(fullOrgCode) != -1).findFirst();
                if (first.isPresent()) {
                    ChannelOrg channelOrg1 = orgAllMap.get(orgCode2);
                    if (channelOrg1 != null) {
                        deptId = channelOrg1.getDeptId();
                    } else {
                        logger.info("deptId:" + orgCode2);
                    }
                }
                if (deptId != null) {
                    deptName = deptMap.get(deptId);
                }
                if (CollUtil.isNotEmpty(result.getDevices())) {
                    List<NewChannelResult.AlreadyDealOrgsResult.DevicesResult> devices = result.getDevices();
                    for (NewChannelResult.AlreadyDealOrgsResult.DevicesResult dResult : devices) {
                        Long finalDeptId = deptId;
                        String finalDeptName = deptName;
                        logger.info(channelOrg.getOrgCode() + "size:" + dResult.getChannels() != null ? dResult.getChannels().size() + "" : "0");
                        buildChannel(aMap, map, channelList, date, orgCode, channelOrg, dResult, finalDeptId, finalDeptName);
                    }
                }
            }

            updateDb(channelList);
            countByOrgQty();
            XxlJobHelper.log("拉取点位成功");
        } catch (InterruptedException e) {
            logger.error("拉取点位失败", e);
        }
    }


    private void buildChannel(Map<String, String> aMap, Map<String, String> map, List<Channel> channelList, Date date, String orgCode, ChannelOrg channelOrg, NewChannelResult.AlreadyDealOrgsResult.DevicesResult dResult, Long finalDeptId, String finalDeptName) {
        try {
            List<NewChannelResult.AlreadyDealOrgsResult.DevicesResult.ChannelsResult> channels = dResult.getChannels();
            for (NewChannelResult.AlreadyDealOrgsResult.DevicesResult.ChannelsResult channel : channels) {
                String gpsX = channel.getGpsX();
                if (StrUtil.isBlank(gpsX) || "0.0".equals(gpsX)) {
                    continue;
                }
                Channel item = new Channel();
                item.setChannelCode(channel.getChannelCode());
                item.setChannelName(channel.getChannelName());
                item.setChannelSn(dResult.getSn());
                item.setCameraType(channel.getCameraType());
                item.setProtocol(map.get(dResult.getProtocolType()));
                item.setIpAddress(dResult.getDeviceIp());
                item.setDeptId(finalDeptId);
                item.setDeptName(finalDeptName);
                if (item.getChannelName().indexOf("五期") != -1) {
                    item.setStageTag("五期");
                } else {
                    item.setStageTag("其他");
                }
                item.setGpsX(channel.getGpsX());
                item.setGpsY(channel.getGpsY());
            /*    if(finalDeptId!=null){
                    // item.setInstallAddress(geocodeRegeo(channel.getGpsX(), channel.getGpsY(), "0"));
                }*/
                if (StrUtil.isNotBlank(channel.getCameraFunctions())) {
                    List<String> split = StrUtil.split(channel.getCameraFunctions(), ",");
                    StringBuffer stringBuffer = new StringBuffer();
                    split.forEach(e -> stringBuffer.append(aMap.get(e)).append(","));
                    String s = stringBuffer.toString();
                    s = s.substring(0, s.length() - 1);
                    item.setAbility(s);
                }
                item.setThreeDDataStatus(1);
                item.setDataSource(1);
                item.setSyncTime(date);
                item.setOrgCode(channelOrg.getOrgCode());
                item.setOrgName(channelOrg.getName());
                item.setCreateBy("system");
                //item.setCreateTime(date);
                item.setUpdateBy("system");
                //item.setUpdateTime(date);
                channelList.add(item);
            }
        } catch (Exception e1) {
            e1.printStackTrace();
        }
    }


    @Transactional(rollbackFor = Exception.class)
    public void pullChannelData() {
        try {
            ChannelPageParam param = new ChannelPageParam();
            int size = 200;
            int page = 1;
            param.setPage(page);
            param.setPageSize(size);
            //查询已审核点位
            //param.setIsExamined(1);
            ChannelPageResult channelPageResult = daHuaApiRepository.channelPage(param);
            List<ChannelResult> records = channelPageResult.getResults();
            if (CollUtil.isEmpty(records)) {
                return;
            }
            //遍历查询所有点位
            if (records.size() == size) {
                while (true) {
                    page++;
                    param.setPage(page);
                    ChannelPageResult pageResult = daHuaApiRepository.channelPage(param);
                    if (pageResult == null || CollUtil.isEmpty(pageResult.getResults())) {
                        break;
                    }
                    List<ChannelResult> resultList = pageResult.getResults();
                    //List<ChannelResult> rList = resultList.stream().filter(e -> e.getChannelName().indexOf(STAGE_NAME) != -1).collect(Collectors.toList());
                    records.addAll(resultList);
                    if (size > resultList.size()) {
                        break;
                    }
                }
            }

            List<Channel> channelList = channelConvert.toChannelDevices(records);
            updateDb(channelList);
            countByOrgQty();
        } catch (Exception e) {
            XxlJobHelper.log("执行失败:" + e.getMessage());
        }
    }

    public void countByOrgQty() {
        List<ChannelOrgCountDto> channelOrgCountDtos = channelService.countByOrgQty();
        Map<String, ChannelOrgCountDto> countMap = channelOrgCountDtos.stream().collect(Collectors.toMap(ChannelOrgCountDto::getOrgCode, e -> e));
        List<ChannelOrg> list = channelOrgService.list();
        Date date = new Date();
        list.forEach(e -> {
            e.setChannelQty(0);
            e.setUpdateTime(date);
            e.setOnlineChannelQty(0);
        });
        Map<String, ChannelOrg> orgMap = list.stream().collect(Collectors.toMap(ChannelOrg::getOrgCode, e -> e));
        for (String orgCode : countMap.keySet()) {
            ChannelOrg channelOrg = orgMap.get(orgCode);
            if (channelOrg == null) {
                continue;
            }
            String fullOrgCode = channelOrg.getFullOrgCode();
            int qty = countMap.get(orgCode).getChannelQty();
            int onlineQty = countMap.get(orgCode).getOnlineChannelQty();
            List<String> codeList = StrUtil.split(fullOrgCode, ",");
            for (String code : codeList) {
                if (orgMap.get(code) != null) {
                    int total = orgMap.get(code).getChannelQty() + qty;
                    int onlineTotal = orgMap.get(code).getOnlineChannelQty() + onlineQty;
                    orgMap.get(code).setChannelQty(total);
                    orgMap.get(code).setOnlineChannelQty(onlineTotal);
                }
            }
        }

        List<ChannelOrg> collect = list.stream().filter(e -> e.getChannelQty() > 0).collect(Collectors.toList());
        if (CollUtil.isNotEmpty(collect)) {
            channelOrgService.updateBatchById(collect);
        }
    }

    private void updateDb(List<Channel> channelList) {
        Map<String, Channel> channelMap = channelList.stream().collect(Collectors.toMap(Channel::getChannelCode, item -> item));
        Set<String> channelSet = channelMap.keySet();
        //获取库中数据
        List<Channel> dbChannelList = channelService.lambdaQuery().eq(Channel::getDataSource, 1).
                eq(Channel::getDelFlag, "0").list();
        Map<String, Long> dbChannelMap = new HashMap<>();
        dbChannelList.forEach(e -> {
            dbChannelMap.put(e.getChannelCode(), e.getId());
        });
        Set<String> dbChannelSet = dbChannelMap.keySet();
        //比对数据
        //新增点位列表
        Collection<String> addList = CollUtil.subtract(channelSet, dbChannelSet);
        //更新点位列表
        Collection<String> updateList = CollUtil.intersection(channelSet, dbChannelSet);
        //比对删除数据
        Collection<String> deleteList = CollUtil.subtract(dbChannelSet, channelSet);
        // 入库
        if (CollUtil.isNotEmpty(addList)) {
            List<Channel> insertList = new ArrayList<>(addList.size());
            addList.forEach(channelCode -> {
                Channel channel = channelMap.get(channelCode);
                if (channel.getId() == null) {
                    channel.setId(IdUtil.getSnowflakeNextId());
                }
                insertList.add(channel);
            });
            channelService.saveBatch(insertList, 2000);
            dbChannelMap.putAll(insertList.stream().collect(Collectors.toMap(Channel::getChannelCode, Channel::getId)));
        }
        //更新点位列表
        if (CollUtil.isNotEmpty(updateList)) {
            List<Channel> upeList = new ArrayList<>(updateList.size());
            updateList.forEach(channelCode -> {
                Channel channel = channelMap.get(channelCode);
                channel.setId(dbChannelMap.get(channelCode));
                upeList.add(channel);
            });
            channelService.updateBatchById(upeList, 2000);
        }
        //删除点位
        if (CollUtil.isNotEmpty(deleteList)) {
            List<Long> channelIds = new ArrayList<>(deleteList.size());
            deleteList.forEach(e -> {
                Channel channel = channelMap.get(e);
                if (null != channel) {
                    channelIds.add(channel.getId());
                }
            });
            if (CollUtil.isNotEmpty(channelIds)) {
                UpdateWrapper<Channel> updateWrapper = new UpdateWrapper<>();
                updateWrapper.in("id", channelIds);
                updateWrapper.set("del_flag", "1");
                channelService.update(updateWrapper);
            }
        }
    }
}
